package com.uptown.matrix.stream.builder;

import java.util.List;
import java.util.concurrent.TimeUnit;

import com.alibaba.fastjson.JSONObject;
import com.uptown.matrix.stream.callback.RedisPipelineCallBack;
import io.lettuce.core.LettuceFutures;
import io.lettuce.core.RedisClient;
import com.uptown.matrix.common.task.RedisWriterMeta;
import io.lettuce.core.RedisFuture;
import io.lettuce.core.RedisURI;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.async.RedisAsyncCommands;
import io.lettuce.core.codec.ByteArrayCodec;

public class RedisTaskBuilder {

    public static RedisClient buildRedisSink(RedisWriterMeta redisWriterMeta) {

        RedisURI redisUri = RedisURI.Builder
                .redis(redisWriterMeta.getHost())
                .withPort(Integer.parseInt(redisWriterMeta.getPort()))
                .withPassword(redisWriterMeta.getPassword().toCharArray())
                .build();
        return RedisClient.create(redisUri);
    }

    public static void pipeline(RedisClient redisClient, RedisPipelineCallBack pipelineCallBack) {
        StatefulRedisConnection<byte[], byte[]> connect = redisClient.connect(new ByteArrayCodec());
        RedisAsyncCommands<byte[], byte[]> commands = connect.async();

        List<RedisFuture<?>> futures = pipelineCallBack.invoke(commands);
        // 搜集命令统一发送
        commands.flushCommands();
        LettuceFutures.awaitAll(10, TimeUnit.SECONDS, futures.toArray(new RedisFuture[futures.size()]));
        connect.close();
    }
}
